package rdd01.transform

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object CombineByKey_Test {
  //combineByKey()算子测试：需要三个参数，
  //第一个是将相同key的第一个数据进行格式结构转换
  //第二个表示分区内的计算队则
  //第三个表示分区间的计算规则
  def main(args: Array[String]): Unit = {
    val sparkConf=new SparkConf().setMaster("local[*]").setAppName("combineByKey")
    val sparkcontext=new SparkContext(sparkConf)
    val rdd: RDD[(String, Int)] = sparkcontext.makeRDD(List(("b", 4), ("b", 5), ("b", 2), ("a", 6)),
      2)
    val newRDD: RDD[(String, (Int, Int))] = rdd.combineByKey(
      v => (v, 1),
      (t: (Int, Int), v) => {
        (t._1 + v, t._2 + 1)
      },
      (t1: (Int, Int), t2: (Int, Int)) => {
        (t1._1 + t2._1, t1._2 + t2._2)
      }
    )
    val resultRDD: RDD[(String, Int)] = newRDD.mapValues {
      case (num, cnt) => {
        num / cnt
      }
    }
    resultRDD.collect().foreach(println)
  }

}
